#include "stdio.h"
#include "DYListenThread.h"
#include "DYSvrNetMan.h"


DYListenThread::DYListenThread(void)
{
	m_nWaitTime = DD_WAITTIME;
	m_nTryTimes = 5;


	m_nListenType = 0;
	m_nPort	      = 9000;

	m_pTcpSocketEvent = NULL;
	m_pUdpSocketEvent = NULL;
}

DYListenThread::~DYListenThread(void)
{
	Stop();

}

int DYListenThread::SetSvrNetMan(DYSvrNetMan *pSvrMan)
{
    m_pSvrNetMan = pSvrMan;

	return TG_ERR_SUCCESS;
}

int DYListenThread::SetConnectTime(UINT32 nWaitTime, UINT32 nTryTimes)
{
	m_nWaitTime = nWaitTime;
	m_nTryTimes = nTryTimes;

	return 0;
}

int DYListenThread::SetListen(UINT16 nListenType,UINT16 nPort)
{
	m_nListenType = nListenType;
	m_nPort	      = nPort;
	
	return TG_ERR_SUCCESS;
}



int DYListenThread::Stop()
{
	DDThread::m_bThreadRun  = FALSE;
	DDThread::DDThdUnInit();

	if( NULL != m_pTcpSocketEvent )
	{
		sem_post(m_pTcpSocketEvent);
	}
	if( NULL != m_pUdpSocketEvent )
	{
		sem_post(m_pUdpSocketEvent);
	}

	DYWorkerThreadMan::DYWorkerUnInit();

	if( NULL != m_pTcpSocketEvent )
	{
		sem_destroy( m_pTcpSocketEvent );
		m_pTcpSocketEvent = NULL;
	}
	if( NULL != m_pUdpSocketEvent )
	{
		sem_destroy( m_pUdpSocketEvent );
		m_pUdpSocketEvent = NULL;
	}

	return TG_ERR_SUCCESS;
}

#define TG_SVR_INIT_WAITTIME      6000	 //�����̳߳�ʼ��ʱ�趨ʱ��

int DYListenThread::Start( UINT16 nWorkThreadCount )
{
	Stop();

	do
	{
		if( NULL == m_pSvrNetMan )
		{
			break;
		}

		m_pTcpSocketEvent = &m_tTcpSocketEvent;
		sem_init( m_pTcpSocketEvent,0,0 );
		m_pUdpSocketEvent = &m_tUdpSocketEvent;
		sem_init( m_pUdpSocketEvent,0,0 );
		
		
		if( FALSE == DDThread::DDThdInit() )
		{
			break;
		}

		if( FALSE == DYWorkerThreadMan::DYWorkerInit(nWorkThreadCount) )
		{
			break;
		}
			
		return TG_ERR_SUCCESS;

	}while(false);

	Stop();

	return TG_ERR_FAILURE;
}


//listen for data
int DYListenThread::DDThdDispatch()
{
	//SetThreadPriority(GetCurrentThread() , THREAD_PRIORITY_HIGHEST);

	int nTcp = u_open2( m_nPort );
	if( nTcp <= 0 )
	{
		return -1;
	}
	set_noblock( nTcp );


	int nResult = 0;

	while( DDThread::m_bThreadRun )
	{
		try
		{
			// Block until the socket is readable (with a 5-second timeout):
			fd_set rd_set,close_set;

			const unsigned numFds = nTcp+1;
			struct timeval timeout;
			timeout.tv_sec = 5;
			timeout.tv_usec = 0;

			while( DDThread::m_bThreadRun )
			{
				FD_ZERO(&rd_set);
				FD_ZERO(&close_set);
				FD_SET((unsigned)nTcp, &rd_set);
				FD_SET((unsigned)nTcp, &close_set);

				timeout.tv_sec  = 5;
				timeout.tv_usec = 0;

				nResult = select(numFds, &rd_set, NULL, &close_set, &timeout);

				if( nResult > 0 )
				{
					if( DDThread::m_bThreadRun && FD_ISSET(nTcp, &rd_set) )
					{
						int nClientSocket = u_accept(nTcp,NULL,0);
						if( nClientSocket > 0 )
						{
							AddSocket(nClientSocket,1);
						}
					}
				}
			}
		}
		catch (...)
		{
		}
	}

	DDThread::m_bThreadRun = FALSE;

	return 0;
}





int DYListenThread::AddSocket(SOCKET hSocket,UINT32 nType)
{
	if( 1 == nType ) //tcp
	{
		DD_AUTO_LOCK xLock(&m_SectTcpSocket);

		m_arTcpSocket.push_back(hSocket);

		sem_post(m_pTcpSocketEvent);
	}
	else //udp
	{
		DD_AUTO_LOCK xLock(&m_SectUdpSocket);

		m_arUdpSocket.push_back(hSocket);

		sem_post(m_pUdpSocketEvent);
	}

	return 0;
}

int DYListenThread::GetSocket(UINT32 nType,SOCKET &hSocket)
{
	UINT32 nCount = 0;
	if( 1 == nType ) //tcp
	{
		DD_AUTO_LOCK xLock(&m_SectTcpSocket);
		nCount = (UINT32)m_arTcpSocket.size();
		if( 0 == nCount )
		{
			return -1;
		}

		hSocket =  m_arTcpSocket[0];
		m_arTcpSocket.erase( m_arTcpSocket.begin() );
	}
	else //udp
	{
// 		DD_AUTO_LOCK xLock(&m_SectUdpSocket);
// 
// 		nCount = (UINT32)m_arUdpSocket.size();
// 		if( 0 == nCount )
// 		{
// 			sem_wait(m_pUdpSocketEvent);
// 
// 			return -1;
// 		}
// 
// 		hSocket =  m_arUdpSocket[0];
// 		m_arUdpSocket.erase( m_arUdpSocket.begin() );
	}

	return 0;
}





int DYListenThread::DDListenWorkerDispatch(UINT8 *pMsgBuffer)
{
	//SetThreadPriority( GetCurrentThread() , THREAD_PRIORITY_HIGHEST-1 );

	int    nRet    = 0;
	SOCKET hSocket = NULL;

	while( DDThread::m_bThreadRun && DYWorkerThreadMan::m_bWorking )
	{
		if( DDThread::m_bThreadRun && DYWorkerThreadMan::m_bWorking )
		{
			nRet = GetSocket(1,hSocket);
			if( 0 == nRet )
			{
				GetTcpMsgHead(hSocket,pMsgBuffer);
			} 
			else
			{
				sem_wait(m_pTcpSocketEvent);
			}
		}
	}

	return 0;
}





inline int DYListenThread::GetTcpMsgHead(SOCKET hClientSocket,UINT8 *pMsgBuffer)
{
	NET_PACKET_HEAD xMsgHead;

	do 
	{
		try
		{	
			fd_set rd_set,close_set;

			const unsigned numFds = hClientSocket+1;
			struct timeval timeout;
			timeout.tv_sec = 5;
			timeout.tv_usec = 0;

			FD_ZERO(&rd_set);
			FD_ZERO(&close_set);
			FD_SET((unsigned)hClientSocket, &rd_set);
			FD_SET((unsigned)hClientSocket, &close_set);

			int nResult = select(numFds, &rd_set, NULL, &close_set, &timeout);
			if( nResult <= 0 )
			{
				break;
			}
			if( DDThread::m_bThreadRun && FD_ISSET(hClientSocket, &rd_set) )
			{
				nResult = r_read( hClientSocket,(void*)&xMsgHead,sizeof(NET_PACKET_HEAD) );
				if( sizeof(NET_PACKET_HEAD) != nResult )
				{
					break;
				}
				if( xMsgHead.nFrameSize != xMsgHead.nDataLen || 
					xMsgHead.nPacketNum != 0 ||
					!xMsgHead.CheckCRC() )
				{
					break;
				}

				if( xMsgHead.nFrameSize > 0 )
				{
					nResult = r_read( hClientSocket,(char*)pMsgBuffer,xMsgHead.nFrameSize );
					if( xMsgHead.nFrameSize != nResult )
					{
						break;
					}
				}

				FD_ZERO(&rd_set);
				FD_ZERO(&close_set);

				m_pSvrNetMan->TcpDispatch(m_nListenType,hClientSocket,&xMsgHead,pMsgBuffer);

				
				return TG_ERR_SUCCESS;
			}	
		}
		catch(...)
		{
		}

	} while(false);

	r_close( hClientSocket );
	hClientSocket = 0;

	return TG_ERR_FAILURE;
}


int DYListenThread::GetUdpMsgHead(SOCKET hSocket,UINT8 *pMsgBuffer)
{
	
	return TG_ERR_FAILURE;
}











